Python利用Minio API实现文件上传下载、删除等操作 |
您所在的位置：网站首页 › minio browser什么意思 › Python利用Minio API实现文件上传下载、删除等操作 |
Python利用Minio API实现文件上传下载、删除等操作
|
上一篇：Windows下Minio介绍、安装及使用、密码修改 1、 pip安装minio pip3 install minio 2、文件目录
ImportError: cannot import name 'Minio'
pip3 install minio 还是报错ImportError错误。 原因是建的文件夹名与pytnon文件minio包重名了，qaq。。。 3、连接Minio文件存储服务器 #!/usr/bin/python # -*- coding: UTF-8 -*- """ @author:ZSW @file:MinioConfig.py @time:2020/12/04 """ from minio import Minio # 使用endpoint、access key和secret key来初始化minioClient对象。 minioClient = Minio('192.168.18.26:9000', access_key='minioadmin', secret_key='minioadmin', secure=False) 4、Python操作存储桶(文件夹) #!/usr/bin/python # -*- coding: UTF-8 -*- """ @author:ZSW @file:MinioBucketBase.py @time:2020/12/04 """ from minio_learn.MinioConfig import minioClient from minio.error import ResponseError class Bucket: # 创建桶(调用make_bucket来创建一个存储桶) make_bucket函数 """ 注：创建桶命名限制：小写字母，句点，连字符和数字是 唯一允许使用的字符（使用大写字母、下划线等命名会报错），长度至少应为3个字符 """ def create_bucket(self): try: if minioClient.bucket_exists(bucket_name='pictures'): # bucket_exists：检查桶是否存在 print("该存储桶已经存在") else: minioClient.make_bucket("pictures") print("存储桶创建成功") except ResponseError as err: print(err) # 列出所有的存储桶 list_buckets函数 def get_bucket_list(self): try: buckets = minioClient.list_buckets() for bucket in buckets: print(bucket.name, bucket.creation_date) # 获取桶的名称和创建时间 except ResponseError as err: print(err) # 删除存储桶 def get_remove_bucket(self): try: minioClient.remove_bucket("pictures") print("删除存储桶成功") except ResponseError as err: print(err) # 列出存储桶中所有对象 或者使用 list_objects_v2也可 def get_bucket_files(self): try: objects = minioClient.list_objects('testfiles', prefix=None, recursive=True) # prefix用于过滤的对象名称前缀 for obj in objects: print(obj.bucket_name, obj.object_name.encode('utf-8'), obj.last_modified, obj.etag, obj.size, obj.content_type) except ResponseError as err: print(err) # 列出存储桶中未完整上传的对象 def get_list_incomplete_uploads(self): try: uploads = minioClient.list_incomplete_uploads('testfiles', prefix=None, recursive=True) for obj in uploads: print(obj.bucket_name, obj.object_name, obj.upload_id, obj.size) except ResponseError as err: print(err) # 获取存储桶的当前策略 def bucket_policy(self): try: policy = minioClient.get_bucket_policy('testfiles') print(policy) except ResponseError as err: print(err) # # 给指定的存储桶设置存储桶策略 # def get_set_bucket_policy(self): # try: # minioClient.set_bucket_policy('testfiles', policy.READ_ONLY) # except ResponseError as err: # print(err) # 获取存储桶上的通知配置 def bucket_notification(self): try: # 获取存储桶的通知配置。 notification = minioClient.get_bucket_notification('testfiles') print(notification) # 如果存储桶上没有任何通知： # notification == {} except ResponseError as err: print(err) # 给存储桶设置通知配置 def set_bucket_notification(bucket_name, notification): pass # 删除存储桶上配置的所有通知 def remove_all_bucket_notifications(bucket_name): try: minioClient.remove_all_bucket_notifications('mybucket') except ResponseError as err: print(err) # 监听存储桶上的通知 def listen_bucket_notification(bucket_name, prefix, suffix, events): pass if __name__ == '__main__': Bucket().bucket_notification() 5、Python操作文件对象 #!/usr/bin/python # -*- coding: UTF-8 -*- """ @author:ZSW @file:MinioObjectBase.py @time:2020/12/04 """ from minio_learn.MinioConfig import minioClient from minio.error import ResponseError import os class Object: # 从桶中下载一个对象txt、csv文件都可以 def load_object(self): try: data = minioClient.get_object('testfiles', 'long_lat.csv') with open('./load_files/long_lat.csv', 'wb') as file_data: for d in data.stream(32 * 1024): file_data.write(d) print("Sussess") except ResponseError as err: print(err) # 下载一个对象的指定区间的字节数组 def load_partial_object(self): try: data = minioClient.get_partial_object('testfiles', '123.txt', 2, 8) with open('./load_files/123_new.txt', 'wb') as file_data: for d in data: file_data.write(d) print("Sussess") # 部分出现乱码 except ResponseError as err: print(err) # 下载并将文件保存到本地 def fget_object(self): try: print(minioClient.fget_object('testfiles', '123.txt', './load_files/123_new1.txt')) except ResponseError as err: print(err) # 拷贝对象存储服务上的源对象到一个新对象 # 注：该API支持的最大文件大小是5GB # 可通过copy_conditions参数设置copy条件 # 经测试copy复制28M的文件需要663ms; 1.8G的压缩包需要53s def get_copy_object(self): try: copy_result = minioClient.copy_object("pictures", "123.jpg", "/testfiles/123.jpg" ) print(copy_result) except ResponseError as err: print(err) # 添加一个新的对象到对象存储服务 """ 单个对象的最大大小限制在5TB。put_object在对象大于5MiB时，自动使用multiple parts方式上传。 这样，当上传失败时，客户端只需要上传未成功的部分即可（类似断点上传）。 上传的对象使用MD5SUM签名进行完整性验证。 """ def upload_object(self): # 放置一个具有默认内容类型的文件，成功后将打印服务器计算出的etag标识符 try: with open('./picture_files/123.txt', 'rb') as file_data: file_stat = os.stat('./picture_files/123.txt') print(minioClient.put_object('pictures', '123.txt', file_data, file_stat.st_size)) print("Sussess") except ResponseError as err: print(err) # 放一个文件'application/csv' try: with open('./picture_files/long_lat.csv', 'rb') as file_data: file_stat = os.stat('./picture_files/long_lat.csv') minioClient.put_object('pictures', 'long_lat.csv', file_data, file_stat.st_size, content_type='application/csv') print("Sussess") except ResponseError as err: print(err) # 通过文件上传到对象中 def fput_object(self): try: print(minioClient.fput_object('pictures', '234.jpg', './picture_files/234.jpg')) print("Sussess") except ResponseError as err: print(err) try: print(minioClient.fput_object('pictures', 'long_lat.csv', './picture_files/long_lat.csv', content_type='application/csv')) print("Sussess") except ResponseError as err: print(err) # 获取对象的元数据 def stat_object(self): try: print(minioClient.stat_object('pictures', '123.txt')) except ResponseError as err: print(err) # 删除对象 def remove_object(self): try: minioClient.remove_object('pictures', '234.jpg') print("Sussess") except ResponseError as err: print(err) # 删除存储桶中的多个对象 def remove_objects(self): try: objects_to_delete = ['123.txt', 'long_lat.csv'] for del_err in minioClient.remove_objects('testfiles', objects_to_delete): print("Deletion Error: {}".format(del_err)) print("Sussess") except ResponseError as err: print(err) # 删除一个未完整上传的对象 def remove_incomplete_upload(self): try: minioClient.remove_incomplete_upload('testfiles', '123.jpg') print("Sussess") except ResponseError as err: print(err) if __name__ == '__main__': Object().remove_incomplete_upload() 6、Python Presigned操作 from minio_learn.MinioConfig import minioClient from minio.error import ResponseError from datetime import timedelta class Presigned: # 生成一个用于HTTP GET操作的presigned URL def presigned_get_object(self): # 预先获得的对象名称的获取对象URL，将在2天后过期 try: print(minioClient.presigned_get_object('testfiles', '123.jpg', expires=timedelta(days=7))) print("Sussess") # 由于内部预定位确实会获得存储桶位置，因此仍然可能出现响应错误 except ResponseError as err: print(err) # 生成一个用于HTTP PUT操作的presigned URL def presigned_put_object(self): try: print(minioClient.presigned_put_object('testfiles', '123.txt', expires=timedelta(days=7))) print("Sussess") except ResponseError as err: print(err) # 允许给POST操作的presigned URL设置策略条件。这些策略包括比如， # 接收对象上传的存储桶名称，名称前缀，过期策略 def presigned_post_policy(self, PostPolicy): pass if __name__ == '__main__': Presigned().presigned_put_object()Minio Cookbook 中文版链接：Minio SDKs - Python Client API文档 - 《Minio Cookbook 中文版》 - 书栈网 · BookStack |
【本文地址】
今日新闻 |
推荐新闻 |